package com.zto.demo.boot.rocketmq;

import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * Created by ztowh on 2016-9-22.
 */
@Component
@ConfigurationProperties(prefix = "app")
public class RocketMqConsumer {

	private static Logger logger = LoggerFactory
			.getLogger(RocketMqConsumer.class);

	private String subscriberID;
	private String nameSrvAddress;
	private Map<String, String> subscriberTopicMap;
	private DefaultMQPushConsumer consumer;

	@PostConstruct
	public void startConsume() throws MQClientException {
		System.out.println("boot Customer start....");
		
		// 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例<br>
        // 注意：ConsumerGroupName需要由应用来保证唯一
		consumer = new DefaultMQPushConsumer(subscriberID);
		consumer.setNamesrvAddr(nameSrvAddress);
		
		// 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
        // 如果非第一次启动，那么按照上次消费的位置继续消费
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		for (Map.Entry<String, String> subscirbeTopicEntry : subscriberTopicMap
				.entrySet()) {
			consumer.subscribe(subscirbeTopicEntry.getKey(),
					subscirbeTopicEntry.getValue());
		}

		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(
					List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

				for (MessageExt msg : msgs) {
					logger.info("======================>"
							+ new String(msg.getBody()));
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});

		consumer.start();
		logger.info("消费启动开始");
	}
	
	/**
     * Spring bean destroy-method
     */
	@PreDestroy
    public void destroy() {
		System.out.println("boot Customer stop....");
    	consumer.shutdown();
    }

	public void setSubscriberID(String subscriberID) {
		this.subscriberID = subscriberID;
	}

	public void setSubscriberTopicMap(Map<String, String> subscriberTopicMap) {
		this.subscriberTopicMap = subscriberTopicMap;
	}

	public String getSubscriberID() {
		return subscriberID;
	}

	public String getNameSrvAddress() {
		return nameSrvAddress;
	}

	public void setNameSrvAddress(String nameSrvAddress) {
		this.nameSrvAddress = nameSrvAddress;
	}

	public Map<String, String> getSubscriberTopicMap() {
		return subscriberTopicMap;
	}
}